package com.atguigu.sql.pj

import java.text.DecimalFormat

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._

/**
 * description ：对分组内的数据求 各城市的占比率
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/13 11:45
 * modified By ：
 * version:    : 1.0
 */
object CityRemarkUDAF extends UserDefinedAggregateFunction {

  override def inputSchema: StructType = StructType(StructField("city_name", StringType) :: Nil)

  override def bufferSchema: StructType = StructType(StructField("city_map", MapType(StringType, LongType)) :: StructField("city_total", LongType) :: Nil)

  override def dataType: DataType = StringType

  override def deterministic: Boolean = true

  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = Map[String, Long]()
    buffer(1) = 0L
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (!input.isNullAt(0)) {
      val city_name: String = input.getString(0)

      // 处理每个城市map集合
      val map: collection.Map[String, Long] = buffer.getMap[String, Long](0)
      buffer(0) = map + (city_name -> (map.getOrElse(city_name, 0L) + 1L))

      // 处理总数
      buffer(1) = buffer.getLong(1) + 1L
    }
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    // 处理每个城市map集合
    val map1: collection.Map[String, Long] = buffer1.getMap[String, Long](0)
    val map2: collection.Map[String, Long] = buffer2.getMap[String, Long](0)
    buffer1(0) = map1.foldLeft(map2) {
      case (map, (city_name, city_count)) =>
        map + (city_name -> (map.getOrElse(city_name, 0L) + city_count))
    }

    // 处理总数
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  override def evaluate(buffer: Row): String = {
    // 获取总数
    val total: Long = buffer.getLong(1)

    // 获取各城市数量
    val map: collection.Map[String, Long] = buffer.getMap[String, Long](0)

    // 计算占比率，第一名，第二名
    val mapTolist: List[(String, Long)] = map.toList.sortBy(_._2)(Ordering.Long.reverse).take(2)
    var remarks: List[CityRemark] = mapTolist.map {
      case (k, v) =>
        CityRemark(k, v.toDouble / total)
    }

    // 计算其他的占比率
    //    remarks = remarks :+ CityRemark("其他", remarks.foldLeft(1D)(_ - _.cityRatio))
    remarks = remarks :+ CityRemark("其他", (1D /: remarks) (_ - _.cityRatio))

    // 返回结果
    remarks.mkString(", ")
  }

  case class CityRemark(cityName: String, cityRatio: Double) {
    val formatter = new DecimalFormat("0.00%")
    override def toString: String = s"$cityName:${formatter.format(cityRatio)}"
  }

}
